package cn.doitedu.rtdw.etl;

import cn.doitedu.rtdw.udfs.GetNullUdf;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job07_ad_show_click_olap_base {

    public static void main(String[] args) {

        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建kafka中dwd行为明细topic的映射表
        tenv.executeSql(
                " create table dwd_events_kafka (                                 "+
                        "      session_id  string                                 "+
                        "     ,user_name    string                                "+
                        "     ,event_id     string                                 "+
                        "     ,action_time   bigint                               "+
                        "     ,properties  map<string,string>                     "+
                        "     ,user_id  bigint                                    "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,device_type string                                 "+
                        "     ,release_channel string                             "+
                        "     ,rt as to_timestamp_ltz(action_time,3)              "+
                        "     ,watermark for rt as rt - interval '0' second       "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    " +
                        "     'properties.group.id' = 'doit43-2',                 " +
                        "     'scan.startup.mode' = 'latest-offset',              " +
                        "     'value.format'='json',                              " +
                        "     'value.json.fail-on-missing-field'='false',         " +
                        "     'value.fields-include' = 'EXCEPT_KEY'   )           "
        );


        // 按照  其他维度 + user_id + 创意id ,聚合： 曝光次数、点击次数、转化次数； 虚拟曝光人数，虚拟点击人数，虚拟转化人数
        tenv.createTemporaryFunction("get_null", GetNullUdf.class);
        tenv.executeSql(
                        " CREATE TEMPORARY VIEW agg_res AS                                                                                     "+
                        " WITH tmp as (                                                                                              "+
                        "     SELECT                                                                                         "+
                        "       user_id,                                                                                     "+
                        "     	event_id,                                                                                    "+
                        "     	action_time,                                                                                 "+
                        "     	properties['creative_id'] as creative_id,                                                    "+
                        "     	rt                                                                                           "+
                        "     FROM dwd_events_kafka                                                                          "+
                        "     WHERE event_id in ('ad_show','ad_click','ad_transform')                                        "+
                        " )                                                                                                  "+
                        " SELECT                                                                                             "+
                        "   to_date(date_format(window_start,'yyyy-MM-dd')) as dt,                                         "+
                        " 	creative_id,                                                                                     "+
                        " 	count(1) filter(where event_id = 'ad_show') as show_cnt,                                         "+
                        " 	count(1) filter(where event_id = 'ad_click') as click_cnt,                                       "+
                        " 	count(1) filter(where event_id = 'ad_transform') as transform_cnt,                               "+
                        " 	user_id as show_user_id,                                                                         "+
                        " 	if(count(1) filter(where event_id = 'ad_click')>0,user_id,get_null() ) as click_user_id,         "+
                        " 	if(count(1) filter(where event_id = 'ad_transform')>0,user_id,get_null() ) as transform_user_id  "+
                        " FROM TABLE(                                                                                        "+
                        "     TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '1' MINUTE)                                           "+
                        " )                                                                                                  "+
                        " GROUP BY                                                                                           "+
                        "     window_start,                                                                                  "+
                        " 	window_end,                                                                                      "+
                        " 	user_id,                                                                                         "+
                        " 	creative_id                                                                                      "
        );

        // 关联hbase中的各种广告相关维表


        // 创建doris中的聚合基础表的映射表
        tenv.executeSql(
                " create table ad_show_click_olap_base_doris(                                                   "+
                        "   dt date,                                                                             "+
                        " 	creative_id string,                                                                  "+
                        " 	show_cnt bigint,                                                                     "+
                        " 	click_cnt bigint,                                                                    "+
                        " 	transform_cnt bigint,                                                                "+
                        " 	show_user_id bigint,                                                                 "+
                        " 	click_user_id bigint,                                                                "+
                        " 	transform_user_id bigint                                                             "+
                        " )  with (                                                                              "+
                        "     'connector' = 'doris',                                                             "+
                        "     'fenodes' = 'doitedu:8030',                                                        "+
                        "     'table.identifier' = 'dws.ad_show_click_olap_base',                                "+
                        "     'username' = 'root',                                                               "+
                        "     'password' = 'root',                                                               "+
                        "     'sink.label-prefix' = 'doit42_no6'  ,                                              "+
                        "     'sink.properties.columns' = 'dt,creative_id,show_cnt,click_cnt,transform_cnt,show_user_id,click_user_id, transform_user_id,     "+
                        " 	show_user_bm = to_bitmap(show_user_id),click_user_bm=to_bitmap(click_user_id),       "+
                        " 	transform_user_bm = to_bitmap(transform_user_id)'                                    "+
                        " )                                                                                      "
        );


        // 将结果写入doris
        tenv.executeSql("insert into ad_show_click_olap_base_doris select * from agg_res");

    }

}
